package pers.cz.javacvdemo.redis;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.Limit;
import org.springframework.data.redis.connection.RedisZSetCommands.ZAddArgs;
import org.springframework.data.redis.connection.zset.Aggregate;
import org.springframework.data.redis.connection.zset.DefaultTuple;
import org.springframework.data.redis.connection.zset.Tuple;
import org.springframework.data.redis.connection.zset.Weights;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
import pers.cz.javacvdemo.utils.ReflectUtils;

import java.lang.reflect.InvocationTargetException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author yiyi.su
 * @version 1.0.0
 * @description SortedSetOP
 * @createTime 2022/8/29
 */
@Slf4j
public class SortedSetOp extends Operation {

	private final ZSetOperations<String, String> ZSET;
	private final StringRedisTemplate redisTemplate;

	public SortedSetOp(StringRedisTemplate redisTemplate) {
		this.ZSET = redisTemplate.opsForZSet();
		this.redisTemplate = redisTemplate;
	}

	private byte[] rawKey(String key) {
		byte[] rawKey = null;
		try {
			rawKey = (byte[]) ReflectUtils.invoke(ZSET, "rawKey", key);
		} catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException ignored) {
			rawKey = key.getBytes(StandardCharsets.UTF_8);
		}
		return rawKey;
	}


	private <T> Set<T> deserializeSet(Set<String> set, Class<T> clz) {
		if (set == null) {
			return null;
		}
		return set.stream().map(v -> deserialize(v, clz)).collect(Collectors.toCollection(LinkedHashSet::new));
	}

	private <T> TypedTuple<T> deserializeTuple(TypedTuple<String> tuple, Class<T> clz) {
		if (tuple == null) {
			return null;
		}
		return new DefaultTypedTuple<>(deserialize(tuple.getValue(), clz), tuple.getScore());
	}

	private <T> Set<TypedTuple<T>> deserializeTuples(Set<TypedTuple<String>> tuples, Class<T> clz) {
		if (tuples == null) {
			return null;
		}
		return tuples.stream().map(tup -> new DefaultTypedTuple<>(deserialize(tup.getValue(), clz), tup.getScore()))
				.collect(Collectors.toCollection(LinkedHashSet::new));
	}


	public <T> TypedTuple<T> bZPopMax(String key, long timeout, TimeUnit unit, Class<T> clz) {
		try {
			return deserializeTuple(ZSET.popMax(key, timeout, unit), clz);
		} catch (Exception e) {
			log.warn("bZPopMax exception, key={},timeout={},unit={}", key, timeout, unit, e);
		}
		return null;
	}

	public <T> TypedTuple<T> bZPopMin(String key, long timeout, TimeUnit unit, Class<T> clz) {
		try {
			return deserializeTuple(ZSET.popMin(key, timeout, unit), clz);
		} catch (Exception e) {
			log.warn("bZPopMin exception, key={},timeout={},unit={}", key, timeout, unit, e);
		}
		return null;
	}

	public Boolean zAdd(String key, Object value, double score) {
		try {
			return ZSET.add(key, serialize(value), score);
		} catch (Exception e) {
			log.warn("zAdd exception, key={},value={},score={}", key, value, score, e);
		}
		return null;
	}

	public Boolean zAdd(String key, Object value, double score, ZAddArgs args) {
		try {
			return redisTemplate.execute((RedisCallback<Boolean>) connection -> connection.zSetCommands().zAdd(rawKey(key), score, rawKey(serialize(value)), args));
		} catch (Exception e) {
			log.warn("zAdd exception, key={},value={},score={},args={}", key, value, score, args, e);
		}
		return null;
	}

	public Boolean zAddIfAbsent(String key, Object value, double score) {
		try {
			return zAdd(key, value, score, ZAddArgs.ifNotExists());
		} catch (Exception e) {
			log.warn("zAddIfAbsent exception, key={},value={},score={}", key, value, score, e);
		}
		return null;
	}

	public <T> Long zAdd(String key, Set<TypedTuple<T>> tuples) {
		try {
			Set<TypedTuple<String>> tupleSet = tuples.stream().map(t -> new DefaultTypedTuple<>(serialize(t.getValue()), t.getScore()))
					.collect(Collectors.toSet());
			return ZSET.add(key, tupleSet);
		} catch (Exception e) {
			log.warn("zAdd exception, key={},tuples={}", key, tuples, e);
		}
		return null;
	}

	public <T> Long zAdd(String key, Set<TypedTuple<T>> tuples, ZAddArgs args) {
		try {
			Set<Tuple> tupleSet = tuples.stream().map(t -> new DefaultTuple(rawKey(serialize(t.getValue())), t.getScore())).collect(Collectors.toSet());
			return redisTemplate.execute((RedisCallback<Long>) connection -> connection.zSetCommands().zAdd(rawKey(key), tupleSet, args));
		} catch (Exception e) {
			log.warn("zAdd exception, key={},tuples={},args={}", key, tuples, args, e);
		}
		return null;
	}

	public <T> Long zAddIfAbsent(String key, Set<TypedTuple<T>> tuples) {
		try {
			return zAdd(key, tuples, ZAddArgs.ifNotExists());
		} catch (Exception e) {
			log.warn("zAddIfAbsent exception, key={},tuples={}", key, tuples, e);
		}
		return null;
	}

	public Long zCard(String key) {
		try {
			return ZSET.zCard(key);
		} catch (Exception e) {
			log.warn("zCard exception, key={}", key, e);
		}
		return null;
	}

	public Long zCount(String key, double min, double max) {
		try {
			return ZSET.count(key, min, max);
		} catch (Exception e) {
			log.warn("zCount exception, key={},min={},max={}", key, min, max, e);
		}
		return null;
	}

	public <T> Set<T> zDiff(String key, Collection<String> otherKeys, Class<T> clz) {
		try {
			return deserializeSet(ZSET.difference(key, otherKeys), clz);
		} catch (Exception e) {
			log.warn("zDiff exception, key={},otherKeys={}", key, otherKeys, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zDiffWithScores(String key, Collection<String> otherKeys, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.differenceWithScores(key, otherKeys), clz);
		} catch (Exception e) {
			log.warn("zDiffWithScores exception, key={},otherKeys={}", key, otherKeys, e);
		}
		return null;
	}

	public Long zDiffStore(String key, Collection<String> otherKeys, String destKey) {
		try {
			return ZSET.differenceAndStore(key, otherKeys, destKey);
		} catch (Exception e) {
			log.warn("zDiffStore exception, key={},otherKeys={},destKey={}", key, otherKeys, destKey, e);
		}
		return null;
	}

	public Double zIncrBy(String key, Object value, double delta) {
		try {
			return ZSET.incrementScore(key, serialize(value), delta);
		} catch (Exception e) {
			log.warn("zIncrBy exception, key={},value={},delta={}", key, value, delta, e);
		}
		return null;
	}

	public <T> Set<T> zInter(String key, Collection<String> otherKeys, Class<T> clz) {
		try {
			return deserializeSet(ZSET.intersect(key, otherKeys), clz);
		} catch (Exception e) {
			log.warn("zInter exception, key={},otherKeys={}", key, otherKeys, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zInterWithScores(String key, Collection<String> otherKeys, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.intersectWithScores(key, otherKeys), clz);
		} catch (Exception e) {
			log.warn("zInterWithScores exception, key={},otherKeys={}", key, otherKeys, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zInterWithScores(String key, Collection<String> otherKeys, Aggregate aggregate, Weights weights, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.intersectWithScores(key, otherKeys, aggregate, weights), clz);
		} catch (Exception e) {
			log.warn("zInterWithScores exception, key={},otherKeys={},aggregate={},weights={}", key, otherKeys, aggregate, weights, e);
		}
		return null;
	}

	public Long zInterStore(String key, Collection<String> otherKeys, String destKey) {
		try {
			return ZSET.intersectAndStore(key, otherKeys, destKey);
		} catch (Exception e) {
			log.warn("zInterStore exception, key={},otherKeys={},destKey={}", key, otherKeys, destKey, e);
		}
		return null;
	}

	public Long zInterStore(String key, Collection<String> otherKeys, String destKey, Aggregate aggregate, Weights weights) {
		try {
			return ZSET.intersectAndStore(key, otherKeys, destKey, aggregate, weights);
		} catch (Exception e) {
			log.warn("zInterStore exception, key={},otherKeys={},aggregate={},weights={}", key, otherKeys, aggregate, weights, e);
		}
		return null;
	}

	public Long zLexCount(String key, Range<String> range) {
		try {
			return ZSET.lexCount(key, range);
		} catch (Exception e) {
			log.warn("zLexCount exception, key={},range={}", key, range, e);
		}
		return null;
	}

	public List<Double> zMScore(String key, Object... o) {
		try {
			String[] array = Arrays.stream(o).map(this::serialize).toArray(String[]::new);
			return ZSET.score(key, array);
		} catch (Exception e) {
			log.warn("zMScore exception, key={},o={}", key, o, e);
		}
		return null;
	}

	public <T> TypedTuple<T> zPopMax(String key, Class<T> clz) {
		try {
			return deserializeTuple(ZSET.popMax(key), clz);
		} catch (Exception e) {
			log.warn("zPopMax exception, key={}", key, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zPopMax(String key, long count, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.popMax(key, count), clz);
		} catch (Exception e) {
			log.warn("zPopMax exception, key={},count={}", key, count, e);
		}
		return null;
	}

	public <T> TypedTuple<T> zPopMin(String key, Class<T> clz) {
		try {
			return deserializeTuple(ZSET.popMin(key), clz);
		} catch (Exception e) {
			log.warn("zPopMin exception, key={}", key, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zPopMin(String key, long count, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.popMin(key, count), clz);
		} catch (Exception e) {
			log.warn("zPopMin exception, key={},count={}", key, count, e);
		}
		return null;
	}

	public <T> T zRandMember(String key, Class<T> clz) {
		try {
			return deserialize(ZSET.randomMember(key), clz);
		} catch (Exception e) {
			log.warn("zRandMember exception, key={}", key, e);
		}
		return null;
	}

	public <T> TypedTuple<T> zRandMemberWithScore(String key, Class<T> clz) {
		try {
			return deserializeTuple(ZSET.randomMemberWithScore(key), clz);
		} catch (Exception e) {
			log.warn("zRandMemberWithScore exception, key={}", key, e);
		}
		return null;
	}

	public <T> List<T> zRandMember(String key, long count, Class<T> clz) {
		try {
			List<String> members = null;
			if (count > 0) {
				Set<String> set = ZSET.distinctRandomMembers(key, count);
				if (set != null) {
					members = new ArrayList<>(set);
				}
			} else {
				members = ZSET.randomMembers(key, count);
			}
			if (members == null) {
				return null;
			}
			return members.stream().map(v -> deserialize(v, clz)).collect(Collectors.toList());
		} catch (Exception e) {
			log.warn("zRandMember exception, key={},count={}", key, count, e);
		}
		return null;
	}

	public <T> List<TypedTuple<T>> zRandMemberWithScore(String key, long count, Class<T> clz) {
		try {
			List<TypedTuple<String>> members = null;
			if (count > 0) {
				Set<TypedTuple<String>> set = ZSET.distinctRandomMembersWithScore(key, count);
				if (set != null) {
					members = new ArrayList<>(set);
				}
			} else {
				members = ZSET.randomMembersWithScore(key, count);
			}
			if (members == null) {
				return null;
			}
			return members.stream().map(v -> new DefaultTypedTuple<>(deserialize(v.getValue(), clz), v.getScore())).collect(Collectors.toList());
		} catch (Exception e) {
			log.warn("zRandMemberWithScore exception, key={},count={}", key, count, e);
		}
		return null;
	}

	public <T> Set<T> zRange(String key, long start, long end, Class<T> clz) {
		try {
			return deserializeSet(ZSET.range(key, start, end), clz);
		} catch (Exception e) {
			log.warn("zRange exception, key={},start={},end={}", key, start, end, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zRangeWithScores(String key, long start, long end, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.rangeWithScores(key, start, end), clz);
		} catch (Exception e) {
			log.warn("zRangeWithScores exception, key={},start={},end={}", key, start, end, e);
		}
		return null;
	}

	public <T> Set<T> zRangeByLex(String key, Range<String> range, Limit limit, Class<T> clz) {
		try {
			return deserializeSet(ZSET.rangeByLex(key, range, limit), clz);
		} catch (Exception e) {
			log.warn("zRangeByLex exception, key={},range={},limit={}", key, range, limit, e);
		}
		return null;
	}

	public <T> Set<T> zRangeByScore(String key, double min, double max, Class<T> clz) {
		try {
			return deserializeSet(ZSET.rangeByScore(key, min, max), clz);
		} catch (Exception e) {
			log.warn("zRangeByScore exception, key={},min={},max={}", key, min, max, e);
		}
		return null;
	}

	public <T> Set<T> zRangeByScore(String key, double min, double max, long offset, long count, Class<T> clz) {
		try {
			return deserializeSet(ZSET.rangeByScore(key, min, max, offset, count), clz);
		} catch (Exception e) {
			log.warn("zRangeByScore exception, key={},min={},max={},offset={},count={}", key, min, max, offset, count, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zRangeByScoreWithScores(String key, double min, double max, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.rangeByScoreWithScores(key, min, max), clz);
		} catch (Exception e) {
			log.warn("zRangeByScoreWithScores exception, key={},min={},max={}", key, min, max, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zRangeByScoreWithScores(String key, double min, double max, long offset, long count, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.rangeByScoreWithScores(key, min, max, offset, count), clz);
		} catch (Exception e) {
			log.warn("zRangeByScoreWithScores exception, key={},min={},max={},offset={},count={}", key, min, max, offset, count, e);
		}
		return null;
	}

	public Long zRank(String key, Object o) {
		try {
			return ZSET.rank(key, serialize(o));
		} catch (Exception e) {
			log.warn("zRank exception, key={},o={}", key, o, e);
		}
		return null;
	}

	public Long zRem(String key, Object... values) {
		try {
			return ZSET.remove(key, Arrays.stream(values).map(this::serialize).toArray(String[]::new));
		} catch (Exception e) {
			log.warn("zRem exception, key={},values={}", key, values, e);
		}
		return null;
	}

	public Long zRemRangeByRank(String key, long start, long end) {
		try {
			return ZSET.removeRange(key, start, end);
		} catch (Exception e) {
			log.warn("zRemRangeByRank exception, key={},start={},end={}", key, start, end, e);
		}
		return null;
	}

	public Long zRemRangeByLex(String key, Range range) {
		try {
			return ZSET.removeRangeByLex(key, range);
		} catch (Exception e) {
			log.warn("zRemRangeByLex exception, key={},range={}", key, range, e);
		}
		return null;
	}

	public Long zRemRangeByScore(String key, double min, double max) {
		try {
			return ZSET.removeRangeByScore(key, min, max);
		} catch (Exception e) {
			log.warn("zRemRangeByScore exception, key={},min={},max={}", key, min, max, e);
		}
		return null;
	}

	public <T> Set<T> zRevRange(String key, long start, long end, Class<T> clz) {
		try {
			return deserializeSet(ZSET.reverseRange(key, start, end), clz);
		} catch (Exception e) {
			log.warn("zRevRange exception, key={},start={},end={}", key, start, end, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zRevRangeWithScores(String key, long start, long end, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.reverseRangeWithScores(key, start, end), clz);
		} catch (Exception e) {
			log.warn("zRevRangeWithScores exception, key={},start={},end={}", key, start, end, e);
		}
		return null;
	}

	public <T> Set<T> zRevRangeByLex(String key, Range<String> range, Limit limit, Class<T> clz) {
		try {
			return deserializeSet(ZSET.reverseRangeByLex(key, range, limit), clz);
		} catch (Exception e) {
			log.warn("zRevRangeByLex exception, key={},range={},limit={}", key, range, limit, e);
		}
		return null;
	}

	public <T> Set<T> zRevRangeByScore(String key, double min, double max, Class<T> clz) {
		try {
			return deserializeSet(ZSET.reverseRangeByScore(key, min, max), clz);
		} catch (Exception e) {
			log.warn("zRevRangeByScore exception, key={},min={},max={}", key, min, max, e);
		}
		return null;
	}

	public <T> Set<T> zRevRangeByScore(String key, double min, double max, long offset, long count, Class<T> clz) {
		try {
			return deserializeSet(ZSET.reverseRangeByScore(key, min, max, offset, count), clz);
		} catch (Exception e) {
			log.warn("zRevRangeByScore exception, key={},min={},max={},offset={},count={}", key, min, max, offset, count, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zRevRangeByScoreWithScores(String key, double min, double max, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.reverseRangeByScoreWithScores(key, min, max), clz);
		} catch (Exception e) {
			log.warn("zRevRangeByScoreWithScores exception, key={},min={},max={}", key, min, max, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zRevRangeByScoreWithScores(String key, double min, double max, long offset, long count, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.reverseRangeByScoreWithScores(key, min, max, offset, count), clz);
		} catch (Exception e) {
			log.warn("zRevRangeByScoreWithScores exception, key={},min={},max={},offset={},count={}", key, min, max, offset, count, e);
		}
		return null;
	}

	public Long reverseRank(String key, Object o) {
		try {
			return ZSET.reverseRank(key, serialize(o));
		} catch (Exception e) {
			log.warn("reverseRank exception, key={},o={}", key, o, e);
		}
		return null;
	}

	public <T> Cursor<TypedTuple<T>> zScan(String key, ScanOptions options, Class<T> clz) {
		try {
			Cursor<TypedTuple<String>> scan = ZSET.scan(key, options);
			return new ConvertingCursor<>(scan, (tuple) -> new DefaultTypedTuple<>(deserialize(tuple.getValue(), clz), tuple.getScore()));
		} catch (Exception e) {
			log.warn("zScan exception, key={},options={}", key, options, e);
		}
		return null;
	}

	public Double zScore(String key, Object o) {
		try {
			return ZSET.score(key, serialize(o));
		} catch (Exception e) {
			log.warn("zScore exception, key={},o={}", key, o, e);
		}
		return null;
	}

	public <T> Set<T> zUnion(String key, Collection<String> otherKeys, Class<T> clz) {
		try {
			return deserializeSet(ZSET.union(key, otherKeys), clz);
		} catch (Exception e) {
			log.warn("zUnion exception, key={},otherKeys={}", key, otherKeys, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zUnionWithScores(String key, Collection<String> otherKeys, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.unionWithScores(key, otherKeys), clz);
		} catch (Exception e) {
			log.warn("zUnionWithScores exception, key={},otherKeys={}", key, otherKeys, e);
		}
		return null;
	}

	public <T> Set<TypedTuple<T>> zUnionWithScores(String key, Collection<String> otherKeys, Aggregate aggregate, Weights weights, Class<T> clz) {
		try {
			return deserializeTuples(ZSET.unionWithScores(key, otherKeys, aggregate, weights), clz);
		} catch (Exception e) {
			log.warn("zUnionWithScores exception, key={},otherKeys={},aggregate={},weights={}", key, otherKeys, aggregate, weights, e);
		}
		return null;
	}

	public Long zUnionStore(String key, Collection<String> otherKeys, String destKey) {
		try {
			return ZSET.unionAndStore(key, otherKeys, destKey);
		} catch (Exception e) {
			log.warn("zUnionStore exception, key={},otherKeys={},destKey={}", key, otherKeys, destKey, e);
		}
		return null;
	}


	public Long zUnionStore(String key, Collection<String> otherKeys, String destKey, Aggregate aggregate, Weights weights) {
		try {
			return ZSET.unionAndStore(key, otherKeys, destKey, aggregate, weights);
		} catch (Exception e) {
			log.warn("zUnionStore exception, key={},otherKeys={},destKey={},aggregate={},weights={}", key, otherKeys, destKey, aggregate, weights, e);
		}
		return null;

	}

}
